package com.bj58.mis.bigdata.demo2.roc;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.kylin.measure.hllc.HLLCounter;

import java.io.IOException;
import java.net.URI;
import java.nio.*;
import java.util.Arrays;

public class CubeStatistics {
    public static void main(String[] args) throws IOException, InterruptedException {
        String uri = "/Users/zhangfuxue/works/58集团/ROC/kybot_2018_05_18_13_53_24/metadata/cube_statistics/HUAXIANG_ALL_TEXT/5ba30390-59ff-4d77-a492-c6770b1c0dbe.seq";

        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(URI.create(uri), conf);
        Path path = new Path(uri);

        SequenceFile.Reader reader = null;
        try {
            reader = new SequenceFile.Reader(fs, path, conf);
            Writable key = (Writable)
                    ReflectionUtils.newInstance(reader.getKeyClass(), conf);
            Writable value = (Writable)
                    ReflectionUtils.newInstance(reader.getValueClass(), conf);
            long position = reader.getPosition();

            while (reader.next(key, value)) {
                //同步记录的边界
                String syncSeen = reader.syncSeen() ? "*" : "";
                System.out.printf("[%s:%s]\t%s:\t%s\n", position, syncSeen, key, value);
                BytesWritable bytesValue = (BytesWritable) value;
                System.out.println(value.getClass());
                System.out.println(Arrays.toString(bytesValue.getBytes()));
                System.out.println(Arrays.toString( Arrays.copyOf(bytesValue.getBytes(), bytesValue.getLength())));
                System.out.println(bytesValue.getBytes().length);
                System.out.println(bytesValue.getLength());

                if(Integer.parseInt(key.toString())>0){
                    HLLCounter hhl = new HLLCounter();
                    hhl.readRegisters(ByteBuffer.wrap(bytesValue.getBytes(), 0, bytesValue.getLength()));
                    System.out.println("hhl:"+hhl.toString()+", " + hhl.getRegister());

                }


                position = reader.getPosition(); // beginning of next record
                Thread.sleep(1000);
            }
        } finally {
            IOUtils.closeStream(reader);
        }
    }
}
